feat: support transactional consumers in the Reader - #1446
Open
Darkbat91 wants to merge 2 commits into
Open
Conversation
Hide control batches and, under ReadCommitted, drop records of aborted transactions in messageSetReader. Track the offsets hidden and emptied batches consume so the reader resumes past them. Closes segmentio#1332.
Darkbat91
force-pushed
the
fix/skip-control-batches
branch
from
July 31, 2026 20:28
f6b4ebb to
4e1682c
Compare
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Problem
kafka-go did not implement the client side of Kafka transactions. We found this when building a go service interconnected with NiFi and transactions enabled on the put Kafka processor.
transactional producer sees empty messages interleaved with real data. The
protocol requires clients to hide them; kafka-go already parses the Control
attribute bit into
messagesHeader.v2.attributesand never tests it.IsolationLevel: ReadCommitteddoes nothing (Aborted transactional data included in response when setting read_committed isolation level in reader #1332). The broker boundsthe fetch at the last stable offset and returns an
AbortedTransactionslist; the client filters. kafka-go parsed that list into a variable it never
read.
Change
Ports the Java client's
CompletedFetchalgorithm intomessageSetReader, thesingle funnel
Reader,BatchandConnall pass through.Hidden batches consume offsets, so
Batchtracks the highest one skipped andresumes past it.
No exported API is added.
dialer.goandconsumergroup.goare untouched.ReadUncommittedstays the default, matching Kafka.Notes
Readerand exported newControlMessagetypes; this keeps the filter in the fetch path and adds no API.